package io.renren.controller;

import io.renren.common.exception.RRException;
import io.renren.common.utils.AESUtils;
import io.renren.common.utils.R;
import io.renren.common.utils.RedisUtils;
import io.renren.common.utils.SpringContextUtils;
import io.renren.common.validator.Assert;
import io.renren.entity.MemberEntity;
import io.renren.entity.SendPacketEntity;
import io.renren.entity.TokenEntity;
import io.renren.service.MemberService;
import io.renren.service.TokenService;

import java.io.IOException;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.apache.commons.lang.StringUtils;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

@ServerEndpoint(value = "/wsRoom/{connectToken}")
@Component
public class WebSocketRoom {

	private final static Logger logger = LoggerFactory.getLogger(WebSocketRoom.class);

	public final static String UID = UUID.randomUUID().toString();

	// concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketRoom对象。
	public final static CopyOnWriteArraySet<WebSocketRoom> webSocketSet = new CopyOnWriteArraySet<WebSocketRoom>();

	// 当前服务器在线用户
	private final static AtomicInteger onlineUsers = new AtomicInteger(0);

	// 当前服务器房间在线用户
	private final static Map<Long, AtomicInteger> onlineUsersByRoom = new Hashtable<Long, AtomicInteger>();

	// 与某个客户端的连接会话，需要通过它来给客户端发送数据
	private Session session;

	// 会员信息
	private MemberEntity member;

	private TokenService tokenService;

	private MemberService memberService;

	static {
		// 防止长时间没有消息时websocket被中间件自动断开
		new Timer().schedule(new TimerTask() {
			@Override
			public void run() {
				for (WebSocketRoom item : webSocketSet) {
					if (item.session.isOpen() && item.member != null) {
						synchronized (item.session) {
							try {
								item.session.getAsyncRemote().sendText("{\"code\":0,\"msg\":\"ping\"}");
							} catch (Exception e) {
							}
						}
					}
				}
			}
		}, 10000L, 10000L);
	}

	/**
	 * 连接建立成功调用的方法
	 */
	@OnOpen
	public void onOpen(Session session, @PathParam("connectToken") String connectToken) {
		try {
			if (StringUtils.isBlank(connectToken) || (connectToken = AESUtils.decrypt(connectToken)) == null || !connectToken.split("@")[0].equals(DateTime.now().toString("yyyy-MM-dd"))) {
				session.getBasicRemote().sendText("{\"code\":999,\"msg\":\"与服务器断开连接\"}");
				session.close();
				return;
			}
		} catch (IOException e) {
		}
		session.setMaxIdleTimeout(5000);
		this.session = session;
		webSocketSet.add(this);
		tokenService = SpringContextUtils.getBean("tokenService", TokenService.class);
		memberService = SpringContextUtils.getBean("memberService", MemberService.class);
		logger.info("当前总连接数:" + webSocketSet.size());
	}

	/**
	 * 连接关闭调用的方法
	 */
	@OnClose
	public void onClose() {
		try {
			webSocketSet.remove(this);
			// sub在线人数
			onlineUsers(false);
			logger.debug("断开连接");
		} catch (Exception e) {
			logger.error("IO异常", e);
		}
	}

	@OnError
	public void onError(Throwable e) {
	}

	/**
	 * 收到客户端消息后调用的方法
	 *
	 * @param message
	 *            客户端发送过来的消息
	 */
	@OnMessage
	public void onMessage(String message) {
		logger.info("来自客户端的消息:" + message);
		ObjectMapper mapper = new ObjectMapper();
		try {
			// 用户验证
			String token = mapper.readValue(message, Map.class).get("token").toString();
			Assert.hasText(token, "token无效");

			TokenEntity tokenEntity = tokenService.queryByToken(token);
			if (tokenEntity == null || tokenEntity.getExpireTime().getTime() < System.currentTimeMillis()) {
				throw new RRException("登录失效，请重新登录");
			}
			member = memberService.selectById(tokenEntity.getUserId());
			// add在线人数
			onlineUsers(true);
		} catch (Exception e) {
			try {
				session.getBasicRemote().sendText(mapper.writeValueAsString(R.error(999, e.getMessage())));
				session.close();
			} catch (IOException e1) {
			}
			e.printStackTrace();
		}
	}

	/**
	 * 群发消息
	 * 
	 * @param message
	 * @throws JsonProcessingException
	 */
	public static void broadcast(Long roomId, String msg) {
		for (WebSocketRoom item : webSocketSet) {
			if (item.session.isOpen() && item.member != null && roomId.equals(item.member.getRoomId())) {
				synchronized (item.session) {
					try {
						item.session.getAsyncRemote().sendText(msg);
					} catch (Exception e) {
						logger.error("broadcast消息推送失败", e);
					}
				}
			}
		}
	}

	/**
	 * 群发红包消息
	 * 
	 * @param message
	 * @throws JsonProcessingException
	 */
	public static void broadcastRP(SendPacketEntity sendPacket, String type) {
		try {
			ObjectMapper mapper = new ObjectMapper();
			String json = mapper.writeValueAsString(R.ok().put("type", type).put("packet", sendPacket));
			for (WebSocketRoom item : webSocketSet) {
				if (item.session.isOpen() && item.member != null && sendPacket.getRoomId().equals(item.member.getRoomId())) {
					synchronized (item.session) {
						try {
							item.session.getAsyncRemote().sendText(json);
						} catch (Exception e) {
							logger.error("broadcastRP消息推送失败", e);
						}
					}
				}
			}
			// 通知其他服务器
			broadcastOtherServer(sendPacket.getRoomId(), json);
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 集群时通知其他服务器
	 * 
	 * @param roomId
	 * @param msg
	 */
	@SuppressWarnings("unchecked")
	private static void broadcastOtherServer(Long roomId, String msg) {
		RedisUtils redisUtils = SpringContextUtils.getBean(RedisUtils.class);
		if (redisUtils == null) {
			return;
		}
		List<String> serverList = redisUtils.get("SERVER_LIST", List.class);
		if (serverList == null) {
			return;
		}
		for (String server : serverList) {
			if (UID.equals(server)) {
				continue;
			}
			for (int i = 0; i < 20; i++) {
				if (redisUtils.get(server + "_msg_" + roomId + "_" + i) == null) {
					redisUtils.set(server + "_msg_" + roomId + "_" + i, roomId + "@" + msg, 10);// 10秒后过期
					break;
				}
			}
		}
	}

	/**
	 * 在线用户统计
	 * 
	 * @param roomId
	 * @param flag
	 */

	private void onlineUsers(boolean flag) {
		if (member == null || member.getRoomId() == null) {
			return;
		}
		Long roomId = member.getRoomId();
		// 在线人数
		RedisUtils redisUtils = SpringContextUtils.getBean(RedisUtils.class);
		if (flag) {
			redisUtils.set("ONLINE_USER_" + member.getId(), "YES");
			redisUtils.set("ONLINE_USERS_" + UID, onlineUsers.incrementAndGet());
			AtomicInteger ouByRoom = onlineUsersByRoom.get(roomId);
			if (ouByRoom == null) {
				ouByRoom = new AtomicInteger(0);
				onlineUsersByRoom.put(roomId, ouByRoom);
			}
			redisUtils.set("ONLINE_USERS_" + UID + "_" + roomId, ouByRoom.incrementAndGet());
		} else {
			redisUtils.delete("ONLINE_USER_" + member.getId());
			redisUtils.set("ONLINE_USERS_" + UID, onlineUsers.decrementAndGet());
			AtomicInteger ouByRoom = onlineUsersByRoom.get(roomId);
			if (ouByRoom == null) {
				logger.error("减去在线人数有误");
				return;
			}
			redisUtils.set("ONLINE_USERS_" + UID + "_" + roomId, ouByRoom.decrementAndGet());
		}
	}
}